package com.redstar.interesting.common.starter.redis;

import com.google.common.collect.Lists;
import com.google.common.primitives.Primitives;
import io.lettuce.core.SetArgs;
import io.lettuce.core.api.async.RedisAsyncCommands;
import io.lettuce.core.cluster.api.async.RedisAdvancedClusterAsyncCommands;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.io.Serializable;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;


/**
 * @author lihongxing
 */
@Component
public class RedisServiceImpl implements RedisService {

    private static final Logger log = LoggerFactory.getLogger(RedisServiceImpl.class);

    @Autowired
    private RedisTemplate<String, String> stringRedisTemplate;

    @Autowired
    private RedisTemplate<String, Serializable> jdkRedisTemplate;


    @Override
    public void delete(String key) {
        jdkRedisTemplate.delete(key);
    }

    @Override
    public void delete(Collection keys) {
        jdkRedisTemplate.delete(keys);
    }

    @Override
    public int getInt(String key) {
        try {
            String obj = stringRedisTemplate.opsForValue().get(key);
            return Integer.parseInt(obj);
        } catch (Exception e) {
            log.error(String.format("exception=%s,key=%s", e.getMessage(), key), e);
        }
        return 0;
    }

    @Override
    public void setInt(String key, int value) {
        stringRedisTemplate.opsForValue().set(key, String.valueOf(value));
    }

    @Override
    public void setInt(String key, int value, long seconds) {
        stringRedisTemplate.opsForValue().set(key, String.valueOf(value));
        stringRedisTemplate.expire(key, seconds, TimeUnit.SECONDS);
    }

    @Override
    public int getAndSet(String key, int value) {
        String oldValue = stringRedisTemplate.opsForValue().getAndSet(key, String.valueOf(value));
        assert oldValue != null;
        return Integer.parseInt(oldValue);
    }

    @Override
    public boolean exists(String key) {
        Boolean aBoolean = stringRedisTemplate.hasKey(key);
        //防止拆箱NPE
        return aBoolean != null && aBoolean;
    }

    @Override
    public Long incr(String key) {
        return stringRedisTemplate.opsForValue().increment(key, 1);
    }

    @Override
    public Long decr(String key) {
        return stringRedisTemplate.opsForValue().increment(key, -1);
    }

    @Override
    public Boolean setnx(String key, String value) {
        return stringRedisTemplate.opsForValue().setIfAbsent(key, value);
    }

    @Override
    @SuppressWarnings("unchecked")
    public Boolean setnxUseMilliseconds(String key, String value, long expire) {
        Boolean resultBoolean = null;
        try {
            resultBoolean = stringRedisTemplate.execute((RedisCallback<Boolean>) connection -> {
                Object nativeConnection = connection.getNativeConnection();
                String redisResult = "";
                RedisSerializer<String> stringRedisSerializer = (RedisSerializer<String>) stringRedisTemplate
                        .getKeySerializer();
                //lettuce连接包下序列化键值，否知无法用默认的ByteArrayCodec解析
                byte[] keyByte = stringRedisSerializer.serialize(key);
                byte[] valueByte = stringRedisSerializer.serialize(value);
                // lettuce连接包下 redis 单机模式setnx
                if (nativeConnection instanceof RedisAsyncCommands) {
                    RedisAsyncCommands commands = (RedisAsyncCommands) nativeConnection;
                    //同步方法执行、setnx禁止异步
                    redisResult = commands
                            .getStatefulConnection()
                            .sync()
                            .set(keyByte, valueByte, SetArgs.Builder.nx().px(expire));
                }
                // lettuce连接包下 redis 集群模式setnx
                if (nativeConnection instanceof RedisAdvancedClusterAsyncCommands) {
                    RedisAdvancedClusterAsyncCommands clusterAsyncCommands = (RedisAdvancedClusterAsyncCommands) nativeConnection;
                    redisResult = clusterAsyncCommands
                            .getStatefulConnection()
                            .sync()
                            .set(keyByte, keyByte, SetArgs.Builder.nx().px(expire));
                }
                //返回加锁结果
                return "OK".equalsIgnoreCase(redisResult);
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
        return resultBoolean != null && resultBoolean;
    }


    @Override
    @SuppressWarnings("unchecked")
    public Boolean setnx(String key, String value, long expire) {
        Boolean resultBoolean = null;
        try {
            resultBoolean = stringRedisTemplate.execute((RedisCallback<Boolean>) connection -> {
                Object nativeConnection = connection.getNativeConnection();
                String redisResult = "";
                RedisSerializer<String> stringRedisSerializer = (RedisSerializer<String>) stringRedisTemplate
                        .getKeySerializer();
                //lettuce连接包下序列化键值，否知无法用默认的ByteArrayCodec解析
                byte[] keyByte = stringRedisSerializer.serialize(key);
                byte[] valueByte = stringRedisSerializer.serialize(value);
                // lettuce连接包下 redis 单机模式setnx
                if (nativeConnection instanceof RedisAsyncCommands) {
                    RedisAsyncCommands commands = (RedisAsyncCommands) nativeConnection;
                    //同步方法执行、setnx禁止异步
                    redisResult = commands
                            .getStatefulConnection()
                            .sync()
                            .set(keyByte, valueByte, SetArgs.Builder.nx().ex(expire));
                }
                // lettuce连接包下 redis 集群模式setnx
                if (nativeConnection instanceof RedisAdvancedClusterAsyncCommands) {
                    RedisAdvancedClusterAsyncCommands clusterAsyncCommands = (RedisAdvancedClusterAsyncCommands) nativeConnection;
                    redisResult = clusterAsyncCommands
                            .getStatefulConnection()
                            .sync()
                            .set(keyByte, keyByte, SetArgs.Builder.nx().ex(expire));
                }
                //返回加锁结果
                return "OK".equalsIgnoreCase(redisResult);
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
        return resultBoolean != null && resultBoolean;
    }

    @Override
    public void expire(String key, long seconds) {
        jdkRedisTemplate.expire(key, seconds, TimeUnit.SECONDS);
    }

    @Override
    public void expireAt(String key, Date date) {
        jdkRedisTemplate.expireAt(key, date);
    }

    @Override
    public long getExpire(String key) {
        return stringRedisTemplate.getExpire(key, TimeUnit.SECONDS);
    }

    @Override
    public <T> T getObject(String key, Class<T> cls) {
        try {
            Serializable obj = jdkRedisTemplate.opsForValue().get(key);
            return Primitives.wrap(cls).cast(obj);
        } catch (Exception e) {
            log.error(String.format("exception=%s,key=%s", e.getMessage(), key), e);
        }
        return null;
    }

    @Override
    public void setObject(String key, Object value) {
        setObject(key, value, null);
    }

    @Override
    public void setObject(String key, Object value, Long seconds) {
        if (value == null || !(value instanceof Serializable)) {
            throw new RuntimeException("对象没有实现序列化");
        }
        try {
            jdkRedisTemplate.opsForValue().set(key, (Serializable) value);
            if (seconds != null) {
                jdkRedisTemplate.expire(key, seconds, TimeUnit.SECONDS);
            }
        } catch (Exception e) {
            log.error(String.format("exception=%s,key=%s", e.getMessage(), key), e);
        }
    }

    @Override
    public void setString(String key, String value) {
        setString(key, value, null);
    }

    @Override
    public void setString(String key, String value, Long seconds) {
        stringRedisTemplate.opsForValue().set(key, value);
        if (seconds != null) {
            stringRedisTemplate.expire(key, seconds, TimeUnit.SECONDS);
        }
    }

    @Override
    public String getString(String key) {
        try {
            return stringRedisTemplate.opsForValue().get(key);
        } catch (Exception e) {
            log.error(String.format("exception=%s,key=%s", e.getMessage(), key), e);
        }
        return null;
    }

    @Override
    public Map<String, String> hGetAll(String key) {
        Map<Object, Object> map = stringRedisTemplate.opsForHash().entries(key);
        Map<String, String> result = new HashMap<>();
        for (Map.Entry<Object, Object> entry : map.entrySet()) {
            result.put(entry.getKey().toString(), entry.getValue().toString());
        }
        return result;
    }

    @Override
    public List<String> hKeys(String key) {
        Set<Object> keys = stringRedisTemplate.opsForHash().keys(key);
        if (CollectionUtils.isEmpty(keys)) {
            return Lists.newArrayList();
        }
        return keys.stream().map(String::valueOf).collect(Collectors.toList());
    }

    @Override
    public String hGet(String key, String hashKey) {
        try {
            return (String) stringRedisTemplate.opsForHash().get(key, hashKey);
        } catch (Exception e) {
            log.error(String.format("exception=%s,key=%s,hashkey=%s", e.getMessage(), key, hashKey), e);
        }
        return null;
    }

    @Override
    public <T> T hGet(String key, Object hashKey, Class<T> clazz) {
        try {
            return Primitives.wrap(clazz).cast(jdkRedisTemplate.opsForHash().get(key, hashKey));
        } catch (Exception e) {
            log.error(String.format("exception=%s,key=%s,hashkey=%s", e.getMessage(), key, hashKey), e);
        }
        return null;
    }

    @Override
    public void hSet(String key, Map<String, List<Serializable>> hash) {
        jdkRedisTemplate.opsForHash().putAll(key, hash);
    }

    @Override
    public void hSet(String key, Object hashKey, Serializable hashValue) {
        jdkRedisTemplate.opsForHash().put(key, hashKey, hashValue);
    }

    @Override
    public <T> List<T> hMGet(String key, Collection<Object> hashKeys, Class<T> clazz) {
        List<Object> objects = jdkRedisTemplate.opsForHash().multiGet(key, hashKeys);
        List<T> list = new ArrayList<>();
        for (Object obj : objects) {
            list.add(Primitives.wrap(clazz).cast(obj));
        }
        return list;
    }

    @Override
    public long hIncrBy(String key, String field, long value) {
        return stringRedisTemplate.opsForHash().increment(key, field, value);
    }

    @Override
    public void hDel(String key, String field) {
        stringRedisTemplate.opsForHash().delete(key, field);
    }

    @Override
    public boolean hExists(String key, String field) {
        return stringRedisTemplate.opsForHash().hasKey(key, field);
    }

    @Override
    public <T> Map<String, T> hGetAll(String key, Class<T> clazz) {
        Map<Object, Object> map = jdkRedisTemplate.opsForHash().entries(key);
        Map<String, T> result = new HashMap<>();
        for (Map.Entry<Object, Object> entry : map.entrySet()) {
            result.put(entry.getKey().toString(), Primitives.wrap(clazz).cast(entry.getValue()));
        }
        return result;
    }


    @Override
    public void hSet(String key, String hashKey, String hashValue) {
        try {
            stringRedisTemplate.opsForHash().put(key, hashKey, hashValue);
        } catch (Exception e) {
            log.error(String.format("exception=%s,key=%s,hashkey=%s", e.getMessage(), key, hashKey), e);
        }
    }




}
